package com.hbase.config;

import com.alibaba.fastjson.JSON;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ContextRefreshedEvent;

import java.io.IOException;
import java.text.MessageFormat;
import java.util.Map;
import java.util.concurrent.*;

/**
 * @description: hbase配置
 * ApplicationListener  启动后执行方法onApplicationEvent
 * 可以使用 HbaseTemplate 来操作hbase，这里使用admin
 * @author: zyj
 * @create: 2019-12-12 14:19
 **/

@Configuration
@EnableConfigurationProperties(HBaseConfigProperties.class)
public class HbaseConfig implements ApplicationListener<ContextRefreshedEvent> {

    Logger logger = LoggerFactory.getLogger(this.getClass());
    @Autowired
    @Qualifier("hBaseConfigProperties")
    private HBaseConfigProperties properties;


    private static org.apache.hadoop.conf.Configuration configuration = HBaseConfiguration.create();
    /**设置连接池*/
    private ThreadLocal<Connection> connectionThreadLocal = new ThreadLocal<>();
    private ThreadLocal<Admin> adminThreadLocal = new ThreadLocal<>();
    private Connection connection = null;
    private Admin admin = null;

    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
    }



    /**
     * 加载HBase配置信息
     */
    private void initConfigurationInfo(){
        Map<String, String> hBaseConfigMap = properties.getConfig();
        if(hBaseConfigMap.size() == 0){
            logger.debug(MessageFormat.format("HBase配置信息初始化失败：{0}", JSON.toJSONString(hBaseConfigMap)));
        }else{
            for (Map.Entry<String,String> confEntry : hBaseConfigMap.entrySet()) {
                configuration.set(confEntry.getKey(), confEntry.getValue());
            }
        }
    }

    /**
    * @Description: 初始化hbase资源
    * @Param: []
    * @return: void
    * @Author: zyj
    * @Date: 2019/12/13
    */
    public void initRepository(){
        try{
            initConfigurationInfo();
            connection = connectionThreadLocal.get();
            if(connection == null){
                connection = ConnectionFactory.createConnection(configuration, newFixedThreadPool(20));
                connectionThreadLocal.set(connection);
            }
            initHBaseClientAdmin();
        }catch (Exception e){
            logger.error(MessageFormat.format("创建hBase connection 连接失败：{0}",e));
            e.printStackTrace();
        }finally {
            close(admin,null,null);
        }
    }


    /**
    * @Description:  初始化HBase client admin
    * @Param: []
    * @return: void
    * @Author: zyj
    * @Date: 2019/12/13
    */
    private  void initHBaseClientAdmin(){
        try{
            admin = adminThreadLocal.get();
            if(admin == null && connection != null){
                admin = connection.getAdmin();
                adminThreadLocal.set(admin);
            }else{
                logger.debug(MessageFormat.format("创建hBase connection连接 失败：{0}",connection));
            }
        }catch (Exception e){
            logger.error(MessageFormat.format("初始化hBase client admin 客户端管理失败：错误信息：{0}",e));
        }
    }

    /**
    * @Description:  关闭流
    * @Param: [admin, rs, table]
    * @return: void
    * @Author: zyj
    * @Date: 2019/12/13
    */
    public void close(Admin admin, ResultScanner rs, Table table){
        if(admin != null){
            try {
                admin.close();
            } catch (IOException e) {
                logger.error("关闭Admin失败",e);
            }
        }
        if(rs != null){
            rs.close();
        }
        if(table != null){
            try {
                table.close();
            } catch (IOException e) {
                logger.error("关闭Table失败",e);
            }
        }
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        initRepository();
    }

    public Admin getAdmin() {
        if(admin==null){
            initHBaseClientAdmin();
        }
        return admin;
    }

    public void setAdmin(Admin admin) {
        this.admin = admin;
    }

    public Connection getConnection() {
        if(null==connection){
            initRepository();
        }
        return connection;
    }

    public void setConnection(Connection connection) {
        this.connection = connection;
    }
}
